package skewreduce.framework;

import java.io.IOException;

import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Counter;

public class LoopingMapper<K1,V1,K2,V2> extends Mapper<K1,V1,K2,V2> {
    protected Counter loopIdx;
    private   long    lIdx;
    protected Counter loopI;
    private   ProgressSensor sensor;

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        super.setup(context);
        context.getCounter("skewreduce","MAP_BEGIN").increment(System.currentTimeMillis());
        loopIdx = context.getCounter("skewreduce","IDX");

        sensor = new ProgressSensor(context);
    }

    protected void beginLoop(Context context) {
        beginLoop(context,0);
    }

    protected void beginLoop(Context context,long n) {
        loopIdx.increment(1);
        /*
        ++lIdx;
        long currentTime = System.currentTimeMillis();
        context.getCounter("skewreduce",String.format("B%d",lIdx)).increment(currentTime);
        if ( n > 0 ) {
            context.getCounter("skewreduce",String.format("N%d",lIdx)).increment(n);
        }

        loopI = context.getCounter("skewreduce",String.format("I%d",lIdx));

        context.progress();
        */
        sensor.beginLoop(n);
    }

    protected void incrLoop(Context context) {
        /*
        loopI.increment(1);
        context.progress();
        */
        sensor.incrLoop();
    }

    protected void endLoop(Context context) {
        /*
        long currentTime = System.currentTimeMillis();
        context.getCounter("skewreduce",String.format("E%d",lIdx)).increment(currentTime);
        context.progress();
        */
        sensor.endLoop();
    }

    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        super.cleanup(context);
        sensor.cleanup();
        context.getCounter("skewreduce","MAP_END").increment(System.currentTimeMillis());
    }
}
